#include <sys/types.h>  /* for type definitions */
#include <sys/socket.h> /* for socket API calls */
#include <netinet/in.h> /* for address structs */
#include <arpa/inet.h>  /* for sockaddr_in */
#include <stdio.h>      /* for printf() and fprintf() */
#include <stdlib.h>     /* for atoi() */
#include <string.h>     /* for strlen() */
#include <unistd.h>     /* for close() */
#include <signal.h>
#include <time.h>
#include <assert.h>
#include <pthread.h>


#define MAX_LEN  1024   /* maximum receive string size */
#define MIN_PORT 1024   /* minimum port allowed */
#define MAX_PORT 65535  /* maximum port allowed */

#define MAX_NODE 100
#define NODE_WEB_SERVER 1
#define NODE_LOG_SERVER 2
#define NODE_LOAD_BALANCER 3
#define NODE_GUI 4
//#define TIMEOUT_INTERVAL 10

int TIMEOUT_INTERVAL = 10;	/* in seconds */

typedef struct node_information 
{
	char addr[16];
	int type;				//[WEB SERVER = 1, LOG SERVER = 2]
	int cpu_load, mem_used, mem_free;
	int count_requests;		// counter of dispatched requests
	char last_req_served[30];	// time of last served request
	int status;					// initiazlied => status = 0 
								// alive =>  status = 1 
								// dead => status = 2
	timer_t t_id ;				//POSIX timer
}node_info;

node_info nodes[MAX_NODE];
node_info sorted_nodes[MAX_NODE];


/* initialize node status */
void init_nodes ()
{
	for( int i=0; i<MAX_NODE; i++) {
		nodes[i].status = 0;
		nodes[i].count_requests = 0;
		strcpy (nodes[i].last_req_served, "");
	}
}


/* when a node joins, get the first free slot to store its data structure */
int first_free_slot ()
{
	for(int i=0; i < MAX_NODE; i++) 
		if ( nodes[i].status != 1) {
			memset ( &nodes[i], 0, sizeof(nodes[i]) );
			return i;
		}			

	return -1;	
}


/* search an active node, and return its index in nodes[]  */
int node_search (char* node_ip)
{
	for(int i=0; i < MAX_NODE; i++)
		if ( strcmp (node_ip, nodes[i].addr) == 0 && nodes[i].status == 1) 
			return i;
	
	return -1;
}


/* when first multicast message comes from a node, node_join is called */
void node_join (char* node_ip, int type, int cpu_load, int mem_used, int mem_free )
{
	printf("Node join called for %s\n", node_ip); 


	//copy node information
	int idx = first_free_slot();
	strcpy (nodes[idx].addr, node_ip) ;
	nodes[idx].type = type;
	nodes[idx].cpu_load = cpu_load;
	nodes[idx].mem_used = mem_used;
	nodes[idx].mem_free = mem_free;
	nodes[idx].status = 1;  //this node is "alive"


	/* configuring signal event */
	struct sigevent	se;
	memset ( (void *)&se, 0, sizeof(se) ) ;
	se.sigev_notify = SIGEV_SIGNAL;
	se.sigev_signo  = SIGUSR1;
	se.sigev_value.sival_int  = idx;	//this value distinugishes which node causes a time out


	/*	creating timer with signal event */
	if ( timer_create (CLOCK_REALTIME, &se, &nodes[idx].t_id) != 0 ) 
	{
		printf ("Failed to create timer for node # %d\n", idx );
		exit (1);
	}


	/* setting timeout values */
	struct itimerspec new_val, old_val ;
	new_val.it_value.tv_sec = TIMEOUT_INTERVAL;
	new_val.it_value.tv_nsec = 0 ;
	new_val.it_interval.tv_nsec = 0 ;
	new_val.it_interval.tv_sec = 0;


	/* schedule the timers */
	if ( timer_settime ( nodes[idx].t_id, 0, &new_val, &old_val) != 0 ) 
	{
		printf("Failed to set timer for node # %d\n", idx ) ;
		exit (1);
	}
}


/* when a time out occurs for a timer, timer value is used to identify which node has dead (and failed to send a multicast packet) */
void sig_handler (int sig_no, siginfo_t *info, void *context)
{
	if ( sig_no == SIGUSR1 ) 
		for(int i=0; i < MAX_NODE; i++) 
			if ( info->si_value.sival_int == i ) 
			{ 
				printf("Node %s has failed!\n", nodes[i].addr ) ;
				nodes[i].status = 2;	// node is dead
			}
}

/* comparator for qsort() */
int comparator (const void* a, const void* b)
{
	return ( ((node_info *)b)->cpu_load - ((node_info *)a)->cpu_load );
}

void sorted_active_nodelist (int &count)	
{
	//copy active nodes
	int j = 0;
	for(int i=0; i < MAX_NODE; i++) 
		if ( nodes[i].status == 1 ) 
			sorted_nodes[j++] = nodes[i];

	//sort active nodes in desending order of cpu (idle) load
	qsort ( sorted_nodes, j, sizeof(sorted_nodes[0]), comparator );
	count = j;
}


/* when a request comes to FM (from LB), count_requests & last_req_served is updated for the webserver with highest idle cpu */
void update_node_info (char *node_addr, int k)
{
	char read_line[MAX_LEN];

	system ("date > req.txt" );
	FILE *fp = fopen ( "req.txt", "r");
	fgets (read_line, MAX_LEN, fp);

	assert ( strlen(read_line) <= 30 );

	strcpy ( sorted_nodes[k].last_req_served, read_line );
	sorted_nodes[k].count_requests++;

	for(int i=0; i < MAX_NODE; i++) 
		if ( strcmp (node_addr, nodes[i].addr) == 0 ) 
		{
			strcpy ( nodes[i].last_req_served, sorted_nodes[k].last_req_served );
			nodes[i].count_requests = sorted_nodes[k].count_requests ;
			break;
		}
	
	fclose (fp);
}


unsigned short recv_port;	    /* port to receive request*/
unsigned short send_port;		/* port number to which send a response */

void* handle_requests (void *arg)
{
	int sock;					/* socket to receive request*/
	struct sockaddr_in server_addr;	/* socket address structure to receive request*/				

	/* create a socket for receiving a request (either from LB or from GUI)*/
	if ((sock = socket(PF_INET, SOCK_DGRAM, IPPROTO_UDP)) < 0) 
	{
		perror("sock failed");
		exit(1);
	}	    


	/* bind local server port */
	memset (&server_addr, 0, sizeof(server_addr));
	server_addr.sin_family      = AF_INET;
	server_addr.sin_addr.s_addr = htonl(INADDR_ANY);
	server_addr.sin_port        = htons(recv_port);

	if ((bind(sock, (struct sockaddr *) &server_addr, sizeof(server_addr))) < 0) 
	{
    	perror("bind() failed");
    	exit(1);
  	}


	for ( ; ;)			/* repeatedly asks FM for sending active node list */
	{
		char recv_str[MAX_LEN];     /* string to send */
		unsigned int recv_len;      /* length of string to send */
  		struct sockaddr_in client_addr; /* packet source */
  		unsigned int client_len;        /* source addr length */


		/* clear the receive buffers & structs */
		memset(recv_str, 0, sizeof(recv_str));
		client_len = sizeof(client_addr);
		memset(&client_addr, 0, client_len);


		/* block waiting to receive a packet */
		if ((recv_len = recvfrom(sock, recv_str, MAX_LEN, 0, (struct sockaddr*) &client_addr, &client_len)) < 0) 
		{
			perror("recvfrom() failed");
			exit(1);
		}

		printf("Received %d bytes from %s [port # %u]: %s", recv_len, inet_ntoa(client_addr.sin_addr), ntohs( client_addr.sin_port), recv_str );
		
		
		int active_node_count;
		
		memset (sorted_nodes, 0, sizeof(sorted_nodes) );
		sorted_active_nodelist ( active_node_count );
		

		int type;
		sscanf( recv_str, "%d", &type) ;
		if ( type == NODE_LOAD_BALANCER )				
		{	
			//when a request comes from LB, it's assumeed that LB is going to forward a request to webserver with highest cpu idle		
			for (int k=0; k < MAX_NODE; k++) 
				if ( sorted_nodes[k].type == NODE_WEB_SERVER ) 
				{
					update_node_info ( sorted_nodes[k].addr, k );		//update webserver information 
					break;
				}
		}


		/* sent packet format : 
			 ---------------------------------------------------------------------------------------------------------------------------------------------------
			| node count | - | addr | type | cpu_load | mem_used | mem_free | request count | last request time | --- | ... for each active node ... |  newline |
			 ---------------------------------------------------------------------------------------------------------------------------------------------------
		*/
		char send_str[MAX_LEN];     /* string to send */
		unsigned int send_len;      /* length of string to send */

		memset(send_str, 0, sizeof(send_str));
		sprintf ( send_str, "%d-", active_node_count );

		for( int k = 0; k <active_node_count; k++) 
		{
			char tmp[MAX_LEN] ;
			memset (tmp, 0, sizeof(tmp) );
			sprintf (tmp, "%s,%d,%d,%d,%d,%d,%s", sorted_nodes[k].addr, sorted_nodes[k].type, sorted_nodes[k].cpu_load, 
													sorted_nodes[k].mem_used, sorted_nodes[k].mem_free, sorted_nodes[k].count_requests, 
													sorted_nodes[k].last_req_served);
			strcat( send_str, tmp );	
			strcat( send_str, "---");
		}
		
		strcat (send_str, "\n");
		send_len = strlen(send_str);

		sleep (1);

		if ( sendto(sock, send_str, send_len, 0, (struct sockaddr *) &client_addr, client_len ) != send_len ) 
		{
			perror("sendto() sent incorrect number of bytes");
			exit(1);
		}
	}	
}


int main(int argc, char *argv[]) 
{
	int sock;                     /* socket descriptor */
  	int flag_on = 1;              /* socket option flag */
  	struct sockaddr_in mc_addr;   /* socket address structure */
  	struct ip_mreq mc_req;        /* multicast request structure */
  	char* mc_addr_str;            /* multicast IP address */
  	unsigned short mc_port;       /* multicast port */

  	/* validate number of arguments */
	if (argc != 5) 
	{
    		fprintf(stderr, "Usage: %s <Multicast IP> <Multicast Port>  <Port to listen>  <TIMEOUT_INTERVAL >\n", argv[0]);
    		exit(1);
  	}

  	mc_addr_str = argv[1];      /* arg 1: multicast ip address */
  	mc_port = atoi(argv[2]);    /* arg 2: multicast port number */
	recv_port = atoi (argv[3]);		/* FM listens to recv_port to get request from LB/GUI */
	TIMEOUT_INTERVAL = atoi (argv[4]);	/* timeout interval for each timer */


	/* validate the port range */
  	if ((mc_port < MIN_PORT || mc_port > MAX_PORT) || (recv_port < MIN_PORT || recv_port > MAX_PORT) ) 
	{
    	fprintf(stderr, "Invalid port number argument.\n");
    	fprintf(stderr, "Valid range is between %d and %d.\n",MIN_PORT, MAX_PORT);
    	exit(1);
  	}


	/* initialize nodes[] data sturcture */
  	init_nodes();


	/* launch the thread to respond to Load Balancer and GUI */
	pthread_t	req_server ;
	if ( pthread_create (&req_server, NULL, handle_requests, NULL) != 0 ) 
	{
		perror ("Thread creation failed");
		exit(1);
	}


	/* set up the signal handler for POSIX timers*/
  	struct sigaction sa;
  	sa.sa_sigaction =  &sig_handler ;
  	sigemptyset (&sa.sa_mask);
  	sa.sa_flags = SA_RESTART | SA_SIGINFO;
  	sigaction(SIGUSR1, &sa, NULL );


  	/* create socket to join multicast group on */
  	if ((sock = socket(PF_INET, SOCK_DGRAM, IPPROTO_UDP)) < 0) 
	{
    	perror("socket() failed");
    	exit(1);
  	}


	/* construct a multicast address structure */
  	memset(&mc_addr, 0, sizeof(mc_addr));
  	mc_addr.sin_family      = AF_INET;
  	mc_addr.sin_addr.s_addr = htonl(INADDR_ANY);
  	mc_addr.sin_port        = htons(mc_port);


	/* bind to multicast address to socket */
  	if ((bind(sock, (struct sockaddr *) &mc_addr, sizeof(mc_addr))) < 0) 
	{
    	perror("bind() failed");
    	exit(1);
  	}


	/* construct an IGMP join request structure */
  	mc_req.imr_multiaddr.s_addr = inet_addr(mc_addr_str);
  	mc_req.imr_interface.s_addr = htonl(INADDR_ANY);


	/* send an ADD MEMBERSHIP message via setsockopt */
	if ((setsockopt(sock, IPPROTO_IP, IP_ADD_MEMBERSHIP, &mc_req, sizeof(mc_req))) < 0) 
	{
    	perror("setsockopt() failed");
    	exit(1);
  	}


	for (;;)  /* loop forever */
	{          
  		char recv_str[MAX_LEN+1];     /* buffer to receive string */
  		int recv_len;                 /* length of string received */
  		struct sockaddr_in from_addr; /* packet source */
  		unsigned int from_len;        /* source addr length */


		/* clear the receive buffers & structs */
		memset(recv_str, 0, sizeof(recv_str));
		from_len = sizeof(from_addr);
		memset(&from_addr, 0, from_len);


		/* block waiting to receive a packet */
		if ((recv_len = recvfrom(sock, recv_str, MAX_LEN, 0, (struct sockaddr*)&from_addr, &from_len)) < 0) 
		{
		  perror("recvfrom() failed");
		  exit(1);
		}


		/* output received string */
		char node_ip[16] ;
		strcpy ( node_ip, inet_ntoa(from_addr.sin_addr) );
		printf("Received %d bytes from %s: %s", recv_len, node_ip, recv_str );


		int idx = node_search (node_ip) ;
		int type, cpu_load, mem_used, mem_free; 

		if ( idx == - 1)	/*if this is first packet from node X, node_join (X,...) is called */
		{
			sscanf (recv_str, "%d,%d,%d,%d", &type, &cpu_load, &mem_used, &mem_free) ;
			node_join (node_ip, type, cpu_load, mem_used, mem_free);
		}
		else 
		{	/* for subsequnt packt from node X, timer of node X is reset to next timeout interval */
			struct itimerspec new_val, old_val ;
			new_val.it_value.tv_sec = TIMEOUT_INTERVAL;
			new_val.it_value.tv_nsec = 0 ;
			new_val.it_interval.tv_nsec = 0 ;
			new_val.it_interval.tv_sec = 0;

			if ( timer_settime ( nodes[idx].t_id, 0, &new_val, &old_val) != 0 ) 
			{
				printf("Failed to reset timer for node # %d\n", idx ) ;
				exit (1);
			}

			/* update the value to the nodes[idx] */
			sscanf (recv_str, "%d,%d,%d,%d", &type, &cpu_load, &mem_used, &mem_free) ;
			nodes[idx].cpu_load = cpu_load;
			nodes[idx].mem_used = mem_used;
			nodes[idx].mem_free = mem_free;
		}
	}


	/* send a DROP MEMBERSHIP message via setsockopt */
  	if ((setsockopt(sock, IPPROTO_IP, IP_DROP_MEMBERSHIP, (void*) &mc_req, sizeof(mc_req))) < 0) 
	{
    		perror("setsockopt() failed");
    		exit(1);
  	}

  	close(sock);  
}
